package cn.finder.consumer;

import cn.finder.Config;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;

import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Properties;

// 消费者Demo
// 用在 清算服务系统中
public class ConsumerDemo {
    public static void main(String[] args){

        // 保留之前消费的位置

        final KafkaConsumer<Long, String> consumer = new KafkaConsumer<Long,String>(buildProperties());

        // 撮合成功的交易单 topic_name: matched_order
        consumer.subscribe(Arrays.asList("matched_order"),new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> collection) {
            }
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> collection) {
                //将偏移设置到最开始
                consumer.seekToBeginning(collection);
                // consumer.seek(0,last_pos); // 记得消费的时候从最新位置开始
            }
        });

        System.out.println("清算服务：消费中...");

        while (true) {
            ConsumerRecords<Long, String> records = consumer.poll(100);
            for (ConsumerRecord<Long, String> record : records){
                // 消费
                //  清算业务
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());

                List<OrderInfo> orderInfos = parseOrders(record.value());

                if (orderInfos!=null && orderInfos.size()>0){
                    for(int i=0;i<orderInfos.size();i++){
                        OrderInfo order=orderInfos.get(i);

                        switch (order.getOrder_type()){

                            case OrderInfo.OrderType_Normal:

                                break;

                            case OrderInfo.OrderType_Restoration:
                                break;

                            case OrderInfo.OrderType_Cancel:
                                break;

                            default:
                                break;
                        }

                    }
                }



                // 标记消费位置
                consumer.commitSync();

            }
        }

    }

    private static  List<OrderInfo> parseOrders(String data){
        ObjectMapper mapper = new ObjectMapper();
        try
        {
            List<OrderInfo> orders = mapper.readValue(data,new TypeReference<List<OrderInfo>>() { });
            return orders;
        }
        catch (Throwable e)
        {
            throw new RuntimeException(e);
        }
    }


    private static Properties buildProperties(){
        Properties props = new Properties();

        props.put("bootstrap.servers", Config.Kafka_host);
        props.put("group.id", "test-consumer-group");
        props.put("enable.auto.commit", "false");
        props.put("auto.commit.interval.ms", "1000");

        props.put("key.deserializer", "org.apache.kafka.common.serialization.LongDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        return props;
    }

}
